package com.gitee.sparta.core.process.feature;

import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.helpers.JacksonHelper;
import com.gitee.sparta.common.helpers.RedisKeyHelper;
import com.gitee.sparta.core.bean.SpartaBeanMapping;
import com.gitee.sparta.core.bean.SpartaChannel;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.operate.ReadableTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;

/**
 * 分布式重试特性
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class SpartaDistributedRetryFeature<V> {

    private final SpartaCoreProperties properties;
    private final ReadableTabOperate<V> readableTabOperate;

    public SpartaDistributedRetryFeature(SpartaCoreProperties properties,
                                    ReadableTabOperate<V> readableTabOperate) {
        this.properties = properties;
        this.readableTabOperate = readableTabOperate;
    }

    /**
     * 开始执行重试， 此重试是分布式的重试，防止单机失败的情况出现
     *
     * @param fullTopic         完整的 topic：topic:metaId
     * @param listenTopic       @SpartaListener 内中 topics 描述
     * @param metadata          被深拷贝的 body 对象，防止用户在回调方法中进行修改内容
     * @param channel           用户 ack 等操作的运用
     */
    public void retry(String fullTopic, String listenTopic, Metadata<V> metadata, SpartaChannel channel) {
        // 分段消费时候，不进入重试
        if (metadata.isSegment()) {
            log.warn("[Sparta] Retry segment not support retry, topic: {}, body: {}",
                    listenTopic, JacksonHelper.serialize(metadata.getBody()));
            return;
        }
        // 重试间隔时间不允许小于 0
        if (channel.getRetryIntervalTime() <= 0) {
            channel.setRetryIntervalTime(metadata.getRetryIntervalTime());
        }

        // 超过重试次数限制
        if (metadata.getRetryNumber() >= this.properties.getRetry()) {
            log.warn("[Sparta] Retry more than {} times, no longer retry.", properties.getRenew());
            // 执行删除数据操作
            this.doDelMetadataReadableTab(listenTopic, fullTopic);
            // 执行最终回调操作
            this.doFinalRetry(listenTopic, metadata.getBody());
            return;
        }

        // 执行重试操作
        this.doRetry(metadata, channel, listenTopic, fullTopic);
    }

    /**
     * 开始重试
     *
     * @param metadata      元数据
     * @param channel       用户设置的重试时间
     * @param listenTopic   @SpartaListener 中标识
     * @param fullTopic     完整的topic:   topic:metaId
     */
    private void doRetry(Metadata<V> metadata, SpartaChannel channel, String listenTopic, String fullTopic) {

        final String readableTabKey = RedisKeyHelper.readableKey(this.properties.getSystemName(), this.properties.getIsolation(), listenTopic);
        final String metadataTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME);
        final String timeoutTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(),
                SpartaConstant.TABLE_TIMEOUT_NAME);


        metadata.setRetryNumber(metadata.getRetryNumber() + 1);
        metadata.setExpire(System.currentTimeMillis() + channel.getRetryIntervalTime());
        this.readableTabOperate.renewOrRetry(readableTabKey, timeoutTabKey, metadataTabKey, fullTopic, metadata);
        log.info("[Sparta] Retry topic: {} retry number: {}", listenTopic, metadata.getRetryNumber());
    }

    /**
     * 强制方法加有 @FinalRetry 注解的方法，将参数传递出去
     *
     * @param listenTopic                   topic
     * @param body                          用户自定义的类
     */
    private void doFinalRetry(String listenTopic, V body) {
        // 调用用户提供的最终通知
        SpartaBeanMapping finalRetryMapping = SpartaTaskDispatcherFeature.getFinalRetry(listenTopic);
        try {
            finalRetryMapping.getCallbackMethod().setAccessible(true);
            finalRetryMapping.getCallbackMethod().invoke(finalRetryMapping.getSpringBean(), listenTopic, body);
            finalRetryMapping.getCallbackMethod().setAccessible(false);
        } catch (Exception ex) {
            log.error("[Sparta] Final retry, exception : {}", ex.getMessage(), ex);
        }
    }

    /**
     * 执行删除操作 删除 metadata_tab 与 readable_tab
     *
     * @param listenTopic   @SpartaListener 中 topics 的标注
     * @param fullTopic     完整的 topic:  topic:metaId
     */
    private void doDelMetadataReadableTab (String listenTopic, String fullTopic) {
        // 删除消息 metadata timeout ready
        final String readableTabKey = RedisKeyHelper.readableKey(this.properties.getSystemName(), this.properties.getIsolation(), listenTopic);
        final String metadataTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME);
        this.readableTabOperate.delMetadataReadableTab(fullTopic, readableTabKey, metadataTabKey);
    }

}
